package com.wmh.baseservice.common.redis.mq.core.stream;


import cn.hutool.core.util.TypeUtil;
import cn.hutool.system.SystemUtil;
import com.alibaba.fastjson2.JSON;
import com.wmh.baseservice.common.redis.mq.core.RedisMqTemplate;
import com.wmh.baseservice.common.redis.mq.core.interceptor.RedisMessageInterceptorFactory;
import lombok.Getter;
import lombok.SneakyThrows;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.lang.reflect.Type;

/**
 * redis stream 消息监听抽象类
 *
 * @author 审心
 */
public abstract class AbstractRedisStreamMessageListener<T extends AbstractRedisStreamMessage> implements StreamListener<String, ObjectRecord<String, String>> {


    /**
     * 消息正文类型
     */
    private final Class<T> messageType;

    /**
     * 消息通道
     */
    @Getter
    private final String streamKey;


    @Value("${spring.application.name}")
    @Getter
    private String defGroupName;

    @Getter
    private final String defConsumerName = String.format("%s@%d", SystemUtil.getHostInfo().getAddress(), SystemUtil.getCurrentPID());

    @Resource
    private RedisMessageInterceptorFactory redisMessageInterceptorFactory;

    @Resource
    private RedisMqTemplate redisMqTemplate;

    @SneakyThrows
    protected AbstractRedisStreamMessageListener() {
        this.messageType = getMessageClass();
        this.streamKey = messageType.getDeclaredConstructor().newInstance().getStreamKey();
    }

    /**
     * -- GETTER --
     * 获取消费者对象, 子类可通过此方法 自定义消费者对象
     * 消费组名 默认为 : 默认使用 spring.application.name 应用英明
     * 消费者名 默认为 : 使用本地 IP + 进程编号 (参考自 RocketMQ clientId 的实现)
     */
    public Consumer getConsumer() {
        return Consumer.from(this.defGroupName, this.defConsumerName);
    }

    /**
     * 实例化后自动创建组
     */
    @PostConstruct
    private void createGroup() {
        StringRedisTemplate template = this.redisMqTemplate.getStringRedisTemplate();
        RedisMqTemplate.createGroup(template, this.streamKey, this.getConsumer().getGroup());
    }

    @Override
    public void onMessage(ObjectRecord<String, String> message) {
        T body = JSON.parseObject(message.getValue(), messageType);
        try {
            redisMessageInterceptorFactory.consumeBefore(body);
            this.onMessage(body);
            // 消费完成 ack
            redisMqTemplate.getStringRedisTemplate().opsForStream().acknowledge(this.getConsumer().getGroup(), message);
            redisMessageInterceptorFactory.consumeSucceed(body);
        } finally {
            redisMessageInterceptorFactory.consumeFinally(body);
        }
    }

    public abstract void onMessage(T message);

    /**
     * 通过解析类上的泛型，获得消息类型
     *
     * @return 消息类型
     */
    @SuppressWarnings("unchecked")
    private Class<T> getMessageClass() {
        Type type = TypeUtil.getTypeArgument(getClass(), 0);
        if (type == null) {
            throw new IllegalStateException(String.format("类型(%s) 需要设置消息类型", getClass().getName()));
        }
        return (Class<T>) type;
    }
}
